package com.atuguigu.boot3.core.config;


import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Component;

import java.util.Properties;

@EnableKafka
@Component
@ConditionalOnProperty("spring.kafka.bootstrap-servers")
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String,Object> kafkaListenerContainerFactor(){
        ConcurrentKafkaListenerContainerFactory factory =new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(new ConsumerFactory() {
            @Override
            public Consumer createConsumer(String groupId, String clientIdPrefix, String clientIdSuffix, Properties properties) {
                return null;
            }

            @Override
            public boolean isAutoCommit() {
                return false;
            }
        });
        factory.setConcurrency(3);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

}
